package com.hulk.rpc.transport.client;

import com.hulk.rpc.annotation.HulkService;
import com.hulk.rpc.common.HulkConnectService;
import com.hulk.rpc.common.RemoteContext;
import com.hulk.rpc.config.client.AppConfig;
import com.hulk.rpc.config.common.HostPort;
import com.hulk.rpc.filter.ClientFilter;
import com.hulk.rpc.invoke.Invoker;
import com.hulk.rpc.invoke.InvokerUtils;
import com.hulk.rpc.loadbalance.WeightableAdaptor;
import com.hulk.rpc.param.EmptyMethodParam;
import com.hulk.rpc.param.MethodParam;
import com.hulk.rpc.protocol.Request;
import com.hulk.rpc.protocol.Response;
import com.hulk.rpc.protocol.ResponseStatus;
import com.hulk.rpc.recycle.RecycleUtils;
import com.hulk.rpc.remote.RemoteException;
import com.hulk.rpc.serialize.Serializer;
import com.hulk.rpc.transport.client.future.RequestWithFuture;
import com.hulk.rpc.utils.AtomicUtils;
import com.hulk.rpc.utils.SystemClock;
import com.hulk.rpc.utils.concurrent.ConcurrentIntToObjectArrayMap;
import io.netty.channel.EventLoopGroup;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;

import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerArray;
import java.util.stream.IntStream;

/**
 * @author xuxukang
 * @date 2019-04-08 10:05
 */
@EqualsAndHashCode(callSuper = true)
@Data
@Slf4j
public final class NettyClientContext extends WeightableAdaptor implements Closeable {

  private final AppConfig appConfig;
  public final HostPort serverAddress;

  // 客户端channel数量
  private final int connectCount;
  private final NettyClient nettyClient;
  private final AtomicInteger requestIdSequencer = new AtomicInteger(0);
  private final Semaphore requestSemaphore;
  // 每个channel的连接错误统计
  private final AtomicIntegerArray connectErrorCounter;
  private final int globalTimeout;
  private final List<ClientFilter> clientFilters;
  private final Serializer serializer;
  private final ConcurrentIntToObjectArrayMap<Integer> methodIdToServiceIdMap = new ConcurrentIntToObjectArrayMap<>();
  private final Method heartbeatMethod;
  private final String heartbeatServiceMethodName;

  private volatile Map<String, Integer> serviceMethodNameToServiceIdMap;
  private volatile boolean closed = false;
  private volatile int weight;

  NettyClientContext(EventLoopGroup eventLoopGroup, AppConfig appConfig,
                     List<ClientFilter> clientFilters, HostPort serverAddress) {
    this.appConfig = appConfig;
    this.serverAddress = serverAddress;
    this.clientFilters = clientFilters;
    this.serializer = appConfig.getSerializer();
    this.globalTimeout = appConfig.getGlobalTimeOut();
    this.connectCount = appConfig.getConnectorPerServer();
    this.connectErrorCounter = new AtomicIntegerArray(connectCount);

    this.nettyClient = new NettyClient(serializer, serverAddress, eventLoopGroup, connectCount);
    if (appConfig.getMaxRequestWait() < 1) {
      this.requestSemaphore = null;
    } else {
      this.requestSemaphore = new Semaphore(appConfig.getMaxRequestWait());
    }

    try {
      heartbeatMethod = HulkConnectService.class.getDeclaredMethod("heartbeat");
      heartbeatServiceMethodName = InvokerUtils.getServiceMethodName(appConfig.getGroup(), appConfig.getApp(),
              heartbeatMethod);
    } catch (Exception e) {
      throw new RemoteException("error on init", e);
    }
  }

  @Override
  public int weight() {
    return weight;
  }

  boolean isSupport(String serviceMethodName) {
    if (serviceMethodNameToServiceIdMap == null) {
      return false;
    }

    return serviceMethodNameToServiceIdMap.containsKey(serviceMethodName);
  }

  void initSerializer() throws Exception {
    if (!serializer.isSupportedClassId()) {
      return;
    }

    int serviceId = HulkConnectService.SERVICE_CLASS_ID_REGISTER;
    long timeout = HulkService.default_timeout;

    CompletableFuture<Map<String, Integer>> future = execute(serviceId, timeout);
    Map<String, Integer> classIds = future.get();

    Map<Class<?>, Integer> classIdMap = new HashMap<>();
    classIds.forEach((className, id) -> {
      try {
        classIdMap.put(Class.forName(className), id);
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });

    serializer.setClassIds(classIdMap);
    log.info(serverAddress + " register Serializer.classIds: " + classIdMap);
  }

  boolean heartbeat() {
    return IntStream//
            .range(0, nettyClient.connectCount())//
            .mapToObj(index -> {

              int requestId = AtomicUtils.incrementAndGet(requestIdSequencer);
              Request request = new Request();
              request.setServiceId(HulkConnectService.SERVICE_HEARTBEAT);
              request.setRequestId(requestId);

              CompletableFuture<Response> future = new CompletableFuture<>();

              try {
                if (requestSemaphore != null) {
                  requestSemaphore.acquire();
                }

                boolean allowSend = doRequestFilter(request, heartbeatMethod, heartbeatServiceMethodName);
                if (allowSend) {
                  long expireTime = SystemClock.fast().mills() + HulkService.default_timeout;
                  nettyClient.send(index, new RequestWithFuture(request, future, expireTime));
                } else {
                  future.completeExceptionally(
                          new RemoteException(ClientFilter.CLIENT_FILTER_DENY, false));
                }
              } catch (Exception e) {
                future.completeExceptionally(e);
              }

              return this.<Boolean>handleResult(request, future);
            })//
            .allMatch(future -> {
              try {
                return future.join();
              } catch (Throwable e) {
                if (log.isWarnEnabled()) {
                  log.warn(serverAddress + " heartbeat error", e);
                }
                return false;
              }
            });
  }

  /**
   * 判断客户端是否可用
   */
  boolean isZombie() {
    int sum = 0;
    boolean allZombie = true;

    for (int i = 0; i < connectCount; i++) {
      int error = connectErrorCounter.get(i);
      // 统计整个app的连接的错误总数
      sum += error;

      if (error < appConfig.getConnectErrorThreshold()) {
        // 有一个connect的连接错误数没有达到阈值，allZombie就为false
        allZombie = false;
      }
    }

    return sum >= appConfig.getServerErrorThreshold() || allZombie;
  }

  void clear() {
    methodIdToServiceIdMap.clear();
  }

  void putServiceId(String serviceMethodName, int methodId) {
    Integer serviceId = serviceMethodNameToServiceIdMap.get(serviceMethodName);
    if (serviceId == null) {
      return;
    }

    methodIdToServiceIdMap.put(methodId, serviceId.intValue());
  }

  int getServiceId(int methodId) {
    return methodIdToServiceIdMap.get(methodId);
  }

  <T> CompletableFuture<T> execute(int serviceId, long timeout) {
    return execute(serviceId, timeout, null, null);
  }

  <T> CompletableFuture<T> execute(int serviceId, long timeout, MethodParam methodParam,
                                   Invoker<CompletableFuture<?>> failoverInvoker) {
    if (closed) {
      throw new RemoteException("连接已经关闭");
    }

    int requestId = AtomicUtils.incrementAndGet(requestIdSequencer);
    for (int i = 0; i < connectCount; i++) {
      if (isZombie(channelIndex(requestId))) {
        requestId = AtomicUtils.incrementAndGet(requestIdSequencer);
        continue;
      }
      break;
    }

    Request request = new Request();
    request.setServiceId(serviceId).setRequestId(requestId);
    if (methodParam instanceof EmptyMethodParam) {
      request.setMethodParam(null);
    } else {
      request.setMethodParam(methodParam);
    }

    if (timeout < 0) {
      timeout = globalTimeout;
    }

    CompletableFuture<Response> responseFuture = new CompletableFuture<>();
    try {
      if (requestSemaphore != null) {
        requestSemaphore.acquire();
      }
      boolean allowSend = doRequestFilter(request);
      if (allowSend) {
        long expireTime = SystemClock.fast().mills() + timeout;
        nettyClient.send(channelIndex(request),
                new RequestWithFuture(request, responseFuture, expireTime));
      } else {
        responseFuture.completeExceptionally(new RemoteException(ClientFilter.CLIENT_FILTER_DENY, false));
      }

    } catch (Exception e) {
      responseFuture.completeExceptionally(e);
    }

    if (failoverInvoker == null) {
      return handleResult(request, responseFuture);
    } else {
      return handleResult(request, responseFuture, failoverInvoker, methodParam);
    }
  }

  @SuppressWarnings("unchecked")
  private <T> CompletableFuture<T> handleResult(final Request request,
                                                final CompletableFuture<Response> future) {
    final Method method;
    final String serviceMethodName;
    if (clientFilters.size() == 0) {
      method = null;
      serviceMethodName = null;
    } else {
      method = RemoteContext.getRemoteMethod();
      serviceMethodName = RemoteContext.getServiceMethodName();
    }

    return future.handle((response, throwable) -> {
      if (requestSemaphore != null) {
        requestSemaphore.release();
      }

      boolean error = false;
      if (throwable != null) {
        if (log.isWarnEnabled()) {
          log.warn("request error, requestId: " + request.getRequestId(), throwable);
        }

        error = true;
      }

      if (!error && response == null) {
        String msg = "request error, requestId: " + request.getRequestId();
        if (log.isWarnEnabled()) {
          log.warn(msg);
        }

        error = true;
      }

      if (!error && response.getStatusCode() != ResponseStatus.OK) {
        String msg = " status code is" + response.getStatusCode() + " reason is " + response.getResult();

        if (log.isWarnEnabled()) {
          log.warn(msg);
        }

        error = true;
      }

      doResponseFilter(request, response, method, serviceMethodName, throwable);

      int channelIndex = channelIndex(request);
      if (error) {
        connectErrorCounter.incrementAndGet(channelIndex);
        return null;
      } else {
        connectErrorCounter.set(channelIndex, 0);
        T result = (T) response.getResult();
        RecycleUtils.release(response);

        return result;
      }

    });
  }

  @SuppressWarnings("unchecked")
  private <T> CompletableFuture<T> handleResult(final Request request,
                                                final CompletableFuture<Response> future,
                                                final Invoker<CompletableFuture<?>> failoverInvoker,
                                                final MethodParam methodParam) {

    final Method method;
    final String serviceMethodName;
    if (clientFilters.size() == 0) {
      method = null;
      serviceMethodName = null;
    } else {
      method = RemoteContext.getRemoteMethod();
      serviceMethodName = RemoteContext.getServiceMethodName();
    }

    CompletableFuture<T> futureWithFailover = new CompletableFuture<>();

    // 通过FutureContainer.notifyResponse进行通知
    future.whenComplete((response, throwable) -> {
      if (requestSemaphore != null) {
        requestSemaphore.release();
      }

      boolean error = false;
      if (throwable != null) {
        if (log.isWarnEnabled()) {
          log.warn("request error, requestId: " + request.getRequestId(), throwable);
        }
        error = true;
      }

      if (!error && response == null) {
        if (log.isWarnEnabled()) {
          log.warn("request error, requestId: " + request.getRequestId());
        }
        error = true;
      }

      if (!error && response.getStatusCode() != ResponseStatus.OK) {
        String msg = " status code is" + response.getStatusCode() + " reason is " + response.getResult();
        if (log.isWarnEnabled()) {
          log.warn("request error, requestId: " + request.getRequestId() + msg);
        }
        error = true;
      }

      doResponseFilter(request, response, method, serviceMethodName, throwable);

      int channelIndex = channelIndex(request);
      if (error) {
        if (log.isInfoEnabled()) {
          log.info("远程调用发生错误，使用本地回退方法执行");
        }

        connectErrorCounter.incrementAndGet(channelIndex);

        failoverInvoker.invoke(methodParam).whenComplete((r, t) -> {
          if (t != null) {
            futureWithFailover.completeExceptionally(t);
          } else {
            futureWithFailover.complete((T) r);
          }
        });
      } else {
        connectErrorCounter.set(channelIndex, 0);

        T result = (T) response.getResult();
        RecycleUtils.release(response);

        futureWithFailover.complete(result);
      }
    });

    return futureWithFailover;
  }

  private void doResponseFilter(Request request, Response response, Method method, String serviceMethodName,
                                Throwable throwable) {
    final int filterLength = clientFilters.size();
    if (filterLength == 0) {
      return;
    }

    RemoteContext.setServerAddress(nettyClient.getServerAddress());
    RemoteContext.setClientAddress(nettyClient.getClientAddress());
    RemoteContext.setRemoteMethod(method);
    RemoteContext.setServiceMethodName(serviceMethodName);

    if (response.getStatusCode() == ResponseStatus.OK) {
      for (ClientFilter filter : clientFilters) {
        filter.onReceive(request, response);
      }
    } else {
      for (ClientFilter filter : clientFilters) {
        filter.onError(request, response, throwable);
      }
    }
  }

  private boolean doRequestFilter(Request request) {
    final int filterLength = clientFilters.size();
    if (filterLength == 0) {
      return true;
    }

    // 客户端
    RemoteContext.setServerAddress(nettyClient.getServerAddress());
    RemoteContext.setClientAddress(nettyClient.getClientAddress());

    for (ClientFilter filter : clientFilters) {
      if (!filter.onSend(request)) {
        return false;
      }
    }

    return true;
  }

  private boolean doRequestFilter(Request request, Method method, String serviceMethodName) {
    final int filterLength = clientFilters.size();
    if (filterLength == 0) {
      return true;
    }

    RemoteContext.setServerAddress(nettyClient.getServerAddress());
    RemoteContext.setClientAddress(nettyClient.getClientAddress());
    RemoteContext.setRemoteMethod(method);
    RemoteContext.setServiceMethodName(serviceMethodName);

    for (ClientFilter filter : clientFilters) {
      if (!filter.onSend(request)) {
        return false;
      }
    }

    return true;
  }

  private boolean isZombie(int index) {
    return connectErrorCounter.get(index) >= appConfig.getConnectErrorThreshold();
  }

  private int channelIndex(int requestId) {
    return requestId % connectCount;
  }

  private int channelIndex(Request request) {
    return channelIndex(request.getRequestId());
  }

  @Override
  public void close() throws IOException {
    if (closed) {
      return;
    }

    closed = true;
    nettyClient.close();
  }

  void connect() throws InterruptedException {
    nettyClient.connect();
    // reset
    for (int i = 0; i < connectErrorCounter.length(); i++) {
      connectErrorCounter.set(i, 0);
    }
  }
}
